%% dht服务模块
%% dht中存储了所有的存储服务节点信息
-module(xbt_dht).

-behaviour(gen_server).

%% 本模块对外提供接口：
%% join/0       向所有的dht server注册存储节点
%% find_nodes/2 从dht中找到存储节点信息
%% find_pids/2  从dht中找到存储节点进程信息
%% leave/0      从所有的dht server中移除存储节点
-export([join/0,find_nodes/2,find_pids/2,leave/0]).
%% gen_server回调函数
-export([init/1,handle_call/3,handle_cast/2,
         handle_info/2,terminate/2,code_change/3]).
%% server启动
-export([start_link/0]).

-include("xbt.hrl").

%% 加入到dht网中，成为一个存储服务提供节点
join() ->
    global:trans({?MODULE,self()},
                 fun() ->
                     % 向所有的已经存在的dht进向发出join的请求
                     gen_server:multi_call(?MODULE,{join,self()})
                 end).

%% 从dht中按Name找到Num个存储节点
find_nodes(Name,Num) when is_list(Name),is_integer(Num) ->
    [ N || {N,_} <- find_1(list_to_binary(Name),Num) ];
%% 从dht中按Name找到Num个存储节点
find_nodes(Name,Num) when is_binary(Name),is_integer(Num) ->
    [ N || {N,_} <- find_1(Name,Num) ].
%% 从dht中按Name找到Num个存储节点的Pid
find_pids(Name,Num) when is_list(Name),is_integer(Num) ->
    [ P || {_,P} <- find_1(list_to_binary(Name),Num) ];
%% 从dht中按Name找到Num个存储节点的Pid
find_pids(Name,Num) when is_binary(Name),is_integer(Num) ->
    [ P || {_,P} <- find_1(Name,Num) ].
%% 存储节点从dht网中离开
leave() ->
    global:trans({?MODULE,self()},
                 fun() ->
                     % 向所有的已经存在的dht进向发出unjoin的请求
                     gen_server:multi_call(?MODULE,{unjoin,self()})
                 end).

%% 调用gen_server启动dht服务，这个调用是由supervisor调用的，不由用户直接调用
start_link()           ->
    gen_server:start_link({local,?MODULE},?MODULE,[],[]).

%% 初始化dht服务，gen_server的回调
%% 反回值为ok,state
%% state如果为nil说明本机没有启动存储节点，如果本机启动存储节点后，state会记录下存储服务的pid
init([])    ->
    process_flag(trap_exit,true),
    net_kernel:monitor_nodes(true),
    BootNodes = xbt_config:get_config(bootnodes),
    % 通过ping boot node加入table集群
    enter(BootNodes),
    % 取得当前所有除自己外其它可用的节点
    Nodes = nodes(),
    % 把自己加入的消息告诉所有的已知dht node，同时把现有的node都加到本机的可用node表中
    lists:foreach(fun(N) ->
                            % 向所有其它节点发送有人来了的消息，了解其它节点上是不是有存储节点
                            {?MODULE,N} ! {new_dht,node()},
                            %%%% 模拟new_dht，把自己的信息与其它节点进行一次交换（这步操作应该没有用，回头考虑是不是删除）
                            self() ! {nodeup,N}
                  end, Nodes),
    % 初始化dht数据库，数据库名为xbtdht
    ets:new(xbtdht,[ordered_set,protected,named_table]),
    {ok,nil}.

%% 确认当前节点可以加入到table集群中去，它会通过ping已知的Boot Node，看是否可以连通，以确认加入到了一个table集群中
%% 返回值为可以连接到的BOOT Node的数量
enter(Nodes) when is_list(Nodes) ->
    Num = join_1(Nodes,0),
    if
        Num > 0   ->
            ?DEBUG("[~p]~p boot nodes available~n",[self(),Num]),
            {ok,Num};
        true      ->
            ?DEBUG("[~p]no boot node available~n",[self()]),
            {error,boot}
    end.

terminate(_, State) ->
    ets:delete(xbtdht),
    leave_1(nodes()),
    case State of
        nil                  ->
            ok;
        Pid when is_pid(Pid) ->
            unlink(Pid)
    end.

%% 有存储服务中止了，将这个存储节点从列表中删除
handle_cast({del_member,Node,_Pid},State) ->
    <<NKey:128/big-integer>> = crypto:md5(atom_to_list(Node)),
    ets:delete(xbtdht, NKey),
    {noreply,State};

%% 将Node做MD5成为一个128位的整数，加入到存储节点数据库中去
handle_cast({exchange,Node,Pid},State) ->
    <<NKey:128/big-integer>> = crypto:md5(atom_to_list(Node)),
    ets:insert(xbtdht,{NKey,{Node,Pid}}),
    {noreply,State};

handle_cast(_Msg, State) ->
    {noreply,State}.

%% 当一个存储服务节点发出join请求后把它加入dht存储服务数据库
handle_call({join,Pid},_From,OldState) ->
    FNode = node(Pid),
    if
        % 当join进来的存储节点和dht节点在同一个Node上时，把它们做一个link好让dht监控存储节点的退出消息
        FNode =:= node()       ->
            link(Pid),
            % 当本机的存储节点启动时把state由nil转换为对应的Pid
            State = Pid;
        true                   ->
            State = OldState
    end,
    <<NKey:128/big-integer>> = crypto:md5(atom_to_list(FNode)),
    ets:insert(xbtdht,{NKey,{FNode,Pid}}),
    {reply,ok,State};

%% 当一个存储服务节点发出unjoin请求后把它从dht存储服务数据库中移去
handle_call({unjoin,Pid},_From,OldState) ->
    FNode = node(Pid),
    if
        FNode =:= node()    ->
            unlink(Pid),
            State = nil;
        true                ->
            State = OldState
    end,
    <<NKey:128/big-integer>> = crypto:md5(atom_to_list(FNode)),
    ets:delete(xbtdht,NKey),
    {reply,ok,State};

handle_call(_Msg,_From,State) ->
    {reply,ok,State}.

%% 本节点的存储服务进程退出的消息处理，它向所有的当前已知节点发出del_member消息，请他们把本存储node从他们的列表中清除，同时也把自己从本机的列表中清除
handle_info({'EXIT',Pid1,_},Pid2) when Pid1 =:= Pid2 ->
    gen_server:abcast(nodes(), ?MODULE, {del_member, node(), Pid1}),
    <<NKey:128/big-integer>> = crypto:md5(atom_to_list(node())),
    ets:delete(xbtdht,NKey),
    {noreply,nil};

%% 当一个新dht节点启动时发出nodeup消息，所有已存在的节点都会收到新节点的nodeup消息，如果它们本机运行了存储节点，就会给新启动的节点cast去本机的存储节点信息
handle_info({nodeup,Node}, State) ->
    case State of
        % 如果本节点的State为一个Pid说明本节点有存储节点，向新服务节点说明本节点的存储节点信息
        Pid when is_pid(Pid) ->
            gen_server:cast({?MODULE,Node},{exchange,node(),Pid}),
            {noreply, State};
        _                    ->
            {noreply, State}
    end;

%% 处理一个节点停止的消息，它将直接从dht数据库中删除对应的数据记录
handle_info({nodedown,Node}, State) ->
    <<NKey:128/big-integer>> = crypto:md5(atom_to_list(Node)),
    ets:delete(xbtdht,NKey),
    {noreply,State};

%% 当一个新dht节点启动时发出nodeup消息，所有已存在的节点都会收到新节点的nodeup消息，如果它们本机运行了存储节点，就会给新启动的节点cast去本机的存储节点信息
handle_info({new_dht,Node}, State) ->
    case State of
        % 如果本机有存储节点，向新加入的dht交换自己的存储节点信息
        Pid when is_pid(Pid) ->
            % 向新启动的dht交换自己的存储节点信息
            gen_server:cast({?MODULE,Node},{exchange,node(),Pid}),
            {noreply, State};
        _                        ->
            {noreply, State}
    end;

handle_info(_Info, State) ->
    {noreply, State}.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%% 通过向所有的Boot Node进行ping操作
%% 返回ping通的Boot Node数量
join_1([],Num)           ->
    Num;
join_1([Node|Nodes],Num) ->
    case net_adm:ping(Node) of
        pong  ->
            join_1(Nodes,Num+1);
        pang  ->
            join_1(Nodes,Num)
    end.

%% 从dht中找到Name所存储的存储节点的信息，数据库中的每条记录内容为{Key,{Node,Pid}}，返回的数据为{{Node,Pid},...}
find_1(Name,Num)   ->
    X = ets:info(xbtdht,size),
    if
        % 如果现在已有的存储节点小于等于需要的节点时，反回现在数据库中的所有节点数据
        X =< Num     ->
            [ V || {_,V} <- ets:tab2list(xbtdht)];
        % 通过Name的md5结果值找出偏移后的Num个存储节点
        true         ->
            <<NKey:128/big-integer>> = crypto:md5(Name),
            find_servers(NKey,Num)
    end.

%% 通过Key寻找向后偏移量为Num个数个存储节点信息
find_servers(Key,Num)  ->
    First = ets:first(xbtdht),
    % 尽可能的找出偏移的存储节点
    S1 = find_servers_1(Key,First,Num,[]),
    NS1 = length(S1),
    ?DEBUG("[~p]S1=~p~n",[self(),S1]),
    if
        % 如果找出来的节点的数量不足，重新从头取出足够数量的节点
        NS1 < Num  ->
            S2 = find_servers_2(First,Num-NS1,[]),
            ?DEBUG("[~p]S2=~p~n",[self(),S2]),
            S1 ++ S2;
        true       ->
            S1
    end.

%% 从数据库中尽可能找出Num个数个存储节点来
% 如果找全了Num个，返回找到的存储节点数据
find_servers_1(_,_,0,S) ->
    lists:reverse(S);
% 如果找到头了，返回已经找到的所有存储节点数据
find_servers_1(_,'$end_of_table',_,S) ->
    lists:reverse(S);
% 从Key向后找大于LKey的记录
% LKey:要查的Key的数值
% Key:当前的记录
% Num：还要查的存储节点数量
% S：找到的节点信息数据
find_servers_1(LKey,Key,Num,S) ->
    Key2 = ets:next(xbtdht,Key),
    if
        LKey =< Key   ->
            [{_,V}] = ets:lookup(xbtdht,Key),
            find_servers_1(LKey,Key2,Num-1,[V|S]);
        true          ->
            find_servers_1(LKey,Key2,Num,S)
    end.

%% 从头取出Num个存储节点
find_servers_2(_,0,S)     ->
    lists:reverse(S);
find_servers_2(Key,Num,S)   ->
    [{_,V}] = ets:lookup(xbtdht,Key),
    Key2 = ets:next(xbtdht,Key),
    find_servers_2(Key2,Num-1,[V|S]).

leave_1([])              ->
    ok;
leave_1([Node|Nodes])    ->
    disconnect_node(Node),
    leave_1(Nodes).

